package com.shujia.tour

import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object ScenicTouristWideApp extends Logging {
  def main(args: Array[String]): Unit = {


    if (args.length == 0) {
      log.error("输出参数为空")
      return
    }

    val day_id = args(0)

    log.info(s"当前天时间分区为：$day_id")

    val month_id: String = day_id.substring(0, 6)

    log.info(s"当前月时间分区为：$month_id")

    val spark: SparkSession = SparkSession.builder()
      .appName("CityTouristWideApp")
      .config("spark.sql.shuffle.partitions", "20")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    //景区游客
    val scenic: DataFrame = spark.sql(s"select * from dal_tour.dal_tour_scenic_tourist_msk_d where day_id=$day_id")

    //行政区配置表
    val adminCode: DataFrame = spark.table("dim.dim_admincode")


    //用户画像表
    val userTag: DataFrame = spark.sql(s"select * from dim.dim_usertag_msk_m where month_id=$month_id")


    /**
      *
      * mdn string comment '游客手机号码'
      * ,d_scenic_id string comment '旅游目的地景区代码'
      * ,d_scenic_name string comment '旅游目的地景区名'
      * ,o_city_name string comment '旅游来源地地市名'
      * ,o_province_name string comment '旅游来源地省名'
      * ,d_arrive_time string comment '游客进入景区的时间'
      * ,d_stay_time double comment '游客在该景区停留的时间长度（小时）'
      * ,number_attr string comment '号码归属地'
      * ,gender string comment '性别'
      * ,trmnl_brand string comment '终端品牌'
      * ,pckg_price int comment '套餐'
      * ,conpot int comment '消费潜力'
      * ,age int comment '年龄'
      *
      */

    val wideDF: DataFrame = scenic
      .join(adminCode.hint("broadcast"), $"source_city_id" === $"city_id")
      .join(userTag, "mdn")
      .select(
        $"mdn",
        $"d_scenic_id",
        $"d_scenic_name",
        $"city_name" as "o_city_name",
        $"prov_name" as "o_province_name",
        $"d_arrive_time",
        $"d_stay_time",
        $"number_attr",
        $"gender",
        $"trmnl_brand",
        $"packg" as "pckg_price",
        $"conpot",
        $"age"
      )




    //保存结果
    wideDF.write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dal_tour/dal_tour_scenic_wide_msk_d/day_id=$day_id")



    //增加分区
    spark.sql(s"alter table dal_tour.dal_tour_scenic_wide_msk_d add if not exists partition(day_id='$day_id')")

  }
}
